package com.cskaoyan.mall.order.mq;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.common.constant.RedisConst;
import com.cskaoyan.mall.mq.constant.MqTopicConst;
import com.cskaoyan.mall.order.query.OrderInfoParam;
import com.cskaoyan.mall.order.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RBucket;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * @Author Pudding
 * @Date 2024/7/23 17:48
 * 秒杀消息的消费者
 */
@Slf4j
@Component
public class PromoOrderConsumer {

    @Autowired
    OrderService orderService;
    @Autowired
    RedissonClient redissonClient;

    @PostConstruct
    public void init() throws MQClientException {

        //1. 创建一个消息消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("promo-order-consumer");

        //2. 设置注册中心
        consumer.setNamesrvAddr("47.96.11.99:9876");

        //3. 订阅主题
        consumer.subscribe(MqTopicConst.PROMO_ORDER_TOPIC, "*");

        //4. 设置消息监听器
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                /*
                    在秒杀下单的业务场景中，需要考虑消息的重复消费的情况。
                    因为在秒杀服务中，一个事务消息对应了一次扣减库存，一次扣减库存应该只对应一个订单
                    一个事务消息，也只能生成一个订单，不能生成多个订单
                    所以，在此处，需要保证消息消费的幂等性（消息消费成功了一次，就不能再消费了)
                    如何保证一个消息只被消费成功一次呢？
                    可以在消息成功之前，往Redis中存储一个消费的标记。那么后续就可以依据Redis是否存在这个标记，来判断这个消息是否被消费过了
                 */
                String msgId = msgs.get(0).getMsgId();
                //构建redis的key
                String key = RedisConst.PROMO_RESUME_MESSAGE + msgId;
                //从redis中取出对应的value
                RBucket<Object> bucket = redissonClient.getBucket(key);
                //防止消息被重复消费的标记
                boolean consumeFlag = bucket.trySet("consume");
                if (!consumeFlag){
                    //说明redis中有这个消息消费的标记，说明消息之前已经消费过了
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }


                try {
                    MessageExt messageExt = msgs.get(0);
                    byte[] body = messageExt.getBody();
                    String messageContent = new String(body);

                    //消费消息
                    OrderInfoParam orderInfoParam = JSON.parseObject(messageContent, OrderInfoParam.class);
                    Long orderId = orderService.saveSeckillOrder(orderInfoParam);
                    if (orderId == null){
                        //删除防止重复下单的标记
                        bucket.delete();

                        //返回消息消费失败
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                    //代码走到这里，说明用户的订单已经生成了
                    //在Redis中打一个标记，标记某个用户(userId)购买了某个商品(skuId)的订单已经生成了
                    RMap<String, Long> map = redissonClient.getMap(RedisConst.PROMO_SECKILL_ORDERS);
                    Long userId = orderInfoParam.getUserId();
                    Long skuId = orderInfoParam.getOrderDetailList().get(0).getSkuId();
                    map.put(userId + ":" + skuId, orderId);

                    //返回消费成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

                } catch (Exception e) {
                    e.printStackTrace();
                    //删除防止重复下单的标记
                    bucket.delete();

                    //返回消息消费失败
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });

        //5. 启动消费者
        consumer.start();
        log.info("秒杀订单消费者启动成功...");
    }
}
